package lstore

import (
	"bytes"
	"encoding/binary"
	"fmt"
	"localstore/blob"
	"log"
	"sync"
	"crypto/md5"
)

type LsDoneFn func(retCode int, req *LsReq)

const (
	OpOsd  = (1 << 8)
	OpPg   = (2 << 8)
	OpMeta = (3 << 8)
	OpLog  = (4 << 8)

	OpOsdInit    = 1 | OpOsd
	OpOsdListPgs = 2 | OpOsd
	OpOsdStat    = 3 | OpOsd
	OpOsdRelease = 4 | OpOsd

	OpPgCreate = 5 | OpPg
	OpPgDelete = 6 | OpPg

	OpMetaUpdate = 7 | OpMeta
	OpMetaLoad   = 8 | OpMeta

	OpAppendEntries = 9 | OpLog
	OpPollEntries   = 10 | OpLog // remove those oldest entries
	OpPopEntries    = 11 | OpLog // remove those newest entries
	OpApplyEntries  = 12 | OpLog
)

const (
	CmdCk   = (1 << 8)
	CmdSnap = (2 << 8)

	CmdWriteCk  = 1 | CmdCk
	CmdReadCk   = 2 | CmdCk
	CmdDeleteCk = 3 | CmdCk

	CmdCreateSnap   = 4 | CmdSnap
	CmdDeleteSnap   = 5 | CmdSnap
	cmdRollbackSnap = 6 | CmdSnap
)

const MetaBytes = 128

type MetaCtx struct {
	data [MetaBytes]byte
}

type PgCtx struct {
}

type CkId struct {
	LunId   uint32
	ChunkId uint32
	SnapId  uint32
	PoolId  uint16
	BlkId   uint16
	BlkNum  uint16
	Rsv     uint16
}

type LogCkData struct {
	CkId CkId
	buf  []byte
}

type LogSnapData struct {
	SnapId uint32
}

type LogEntry struct {
	Term  uint64
	Index uint64
	Op    int
	Data  interface{} // *LockCkData or *LogSnapData
}

type LogCtx struct {
	Num     int
	Entries []LogEntry
}

type LsReq struct {
	ReqId  uint64
	OsdId  uint8
	PgId   uint16
	PoolId uint16
	Op     int
	CbDone LsDoneFn
	Ctx    interface{}
}

const NumPgPerOsd = 16384
const NumReplica = 4
const MetaSize = 4096

type LStore struct {
	mu      sync.Mutex
	end     chan int
	q       chan *LsReq
	running bool
	osd     map[int]*Osd
}

func checkMetaMd5(buf []byte) bool {
	h := md5.New()
	h.Write(buf[16:])
	sum := h.Sum(nil)
	for i:=0; i<16; i++ {
		if sum[i] != buf[i] {
			return false
		}
	}
	return true
}

func loadMeta(b *blob.SpdkBlob, off int64) ([]byte, int) {
	buf, ok := blob.SpdkBlobIoRead(b, off, MetaSize*NumReplica)
	if !ok {
		log.Printf("SpdkBlobIoRead error\n")
		return []byte(""), -1
	}
	idx := -1
	maxVer := int64(0)
	ver := int64(0)
	for i:=0; i<NumReplica; i++ {
		pos := i*MetaSize
		ok := checkMetaMd5(buf[pos:pos+MetaSize])
		if ok := Unmarshal(buf[pos+16:pos+24], &ver); !ok {
			log.Fatal("Unmarshal")
			return []byte(""), -1
		}
		if ok && ver > maxVer {
			maxVer = ver
			idx = i
		}
	}
	if idx < 0 {
		log.Printf("Cannot found valid meta block\n")
		return []byte(""), -1
	}
	pos := idx*MetaSize
	return buf[pos:pos+MetaSize], idx
}

func Marshal(arg interface{}) ([]byte, bool) {
	buf := new(bytes.Buffer)
	if err := binary.Write(buf, binary.LittleEndian, arg); err != nil {
		log.Printf("binary.Write, err=%v\n", err)
		return []byte(""), false
	}
	return buf.Bytes(), true
}

func Unmarshal(buf []byte, arg interface{}) bool {
	rd := bytes.NewReader(buf)
	if err := binary.Read(rd, binary.LittleEndian, arg); err != nil {
		log.Println("binary.Read failed:", err)
		return false
	}
	return true
}

func Size(arg interface{}) int {
	size := binary.Size(arg)
	return size
}

func (ls *LStore) osdLoad(req *LsReq) {
	o := &Osd{}
	o.repIdx = 0
	o.id = req.OsdId

	name := fmt.Sprintf("osd%02d", req.OsdId)
	dev := blob.SpdkBsDev(name)
	if o.bs = blob.SpdkBsLoad(dev, nil); o.bs == nil {
		log.Printf("SpdkBsLoad fail\n")
		req.CbDone(-1, req)
		return
	}
	if o.mbId = blob.SpdkBsGetSuper(o.bs); o.mbId<0 {
		log.Printf("SpdkBsGetSuper fail\n")
		req.CbDone(-1, req)
		return
	}
	if o.mb = blob.SpdkBsOpenBlob(o.bs, o.mbId); o.mb == nil {
		log.Printf("SpdkBsOpenBlob fail\n")
		req.CbDone(-1, req)
		return
	}

	buf, idx := loadMeta(o.mb, 0)
	if idx < 0 {
		log.Printf("LoadMeta error\n")
	}
	o.repIdx = (idx + 1) % NumReplica
	o.meta = &OsdMeta{}
	if ok := Unmarshal(buf, o.meta); !ok {
		log.Printf("Unmarshal fail\n")
		req.CbDone(-1, req)
		return
	}

	for w := 0; w < NumPgBmWord; w++ {
		word := o.meta.PgBitmap[w]
		if word == 0 {
			continue
		}
		for i := 0; i < 64; i++ {
			if word & (uint64(1)<<i) != 0 {
				idx := w*64 + i
				o.loadPg(idx)
			}
		}
	}
	ls.osd[int(req.OsdId)] = o
	req.CbDone(0, req)
}

func (ls *LStore) osdInit(req *LsReq) {
	name := fmt.Sprintf("osd%02d", req.OsdId)
	dev := blob.SpdkBsDev(name)
	if blob.SpdkBsExists(dev) {
		ls.osdLoad(req)
		return
	}

	o := &Osd{}
	o.repIdx = 0
	o.id = req.OsdId

	if o.bs = blob.SpdkBsInit(dev, nil); o.bs == nil {
		log.Printf("SpdkBsInit fail\n")
		req.CbDone(-1, req)
		return
	}
	if o.mbId = blob.SpdkBsCreateBlob(o.bs); o.mbId < 0 {
		log.Printf("SpdkBsCreateBlob fail\n")
		req.CbDone(-1, req)
		return
	}
	if ok := blob.SpdkBsSetSuper(o.bs, o.mbId); !ok {
		log.Printf("SpdkBsSetSuper fail\n")
		req.CbDone(-1, req)
		return
	}
	if ok := blob.SpdkBsMdSync(o.bs); !ok {
		log.Printf("SpdkBlobSyncMd fail\n")
		req.CbDone(-1, req)
		return
	}
	if o.mb = blob.SpdkBsOpenBlob(o.bs, o.mbId); o.mb == nil {
		log.Printf("SpdkBsOpenBlob fail\n")
		req.CbDone(-1, req)
		return
	}
	if ok := o.createMeta(); !ok {
		log.Printf("createMeta fail\n")
		req.CbDone(-1, req)
		return
	}
	o.meta = &OsdMeta{}
	if ok := o.syncOsdMeta(); !ok {
		log.Printf("o.syncOsdMeta fail\n")
		req.CbDone(-1, req)
		return
	}

	ls.osd[int(req.OsdId)] = o
	req.CbDone(0, req)
}

func (ls *LStore) SubmitReq(req *LsReq) {
	ls.q <- req
}

func (ls *LStore) getOsd(osdId int) *Osd {
	o, ok := ls.osd[osdId]
	if !ok {
		return nil
	}
	return o
}

func (ls *LStore) dealWithReq(req *LsReq) {
	switch req.Op {
	case OpOsdInit:
		ls.osdInit(req)
	case OpPgCreate:
		ls.pgCreate(req)
	case OpPgDelete:
		ls.pgDelete(req)
	case OpMetaUpdate:
		ls.metaUpdate(req)
	case OpMetaLoad:
		ls.metaLoad(req)
	case OpAppendEntries:
		ls.appendEntries(req)
	case OpPollEntries:
		ls.pollEntries(req)
	case OpPopEntries:
		ls.popEntries(req)
	case OpApplyEntries:
		ls.applyEntries(req)
	}
}

func (ls *LStore) Run() {
	ls.mu.Lock()
	ls.running = true
	for ls.running {
		select {
		case <-ls.end:
			ls.running = false
			break
		case req := <-ls.q:
			ls.dealWithReq(req)
		}
	}
	log.Printf("stopped\n")
	ls.mu.Unlock()
}

func (ls *LStore) Start() {
	ls.mu.Lock()
	if !ls.running {
		go ls.Run()
	}
	ls.mu.Unlock()
}

func (ls *LStore) Stop() {
	ls.end <- 1
}

func NewLStore() (ls *LStore) {
	ls = &LStore{}
	ls.end = make(chan int)
	ls.q = make(chan *LsReq)
	ls.running = false
	ls.osd = make(map[int]*Osd)
	log.SetFlags(log.Lshortfile)
	return ls
}

